package com.ssa;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * ytx Producer
 *
 * @author <a href="mailto:chenlsh@ytx.com">陈录顺</a>
 * @since 1.0.0
 */
public class Producer extends Thread {
    private final KafkaProducer<Integer, String> prod;
    private final String topic;
    private final Boolean isAsync;

    public Producer(String topic, Boolean isAsync) {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.10.151:9092");
        properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty("client.id", "demoProducer");
        prod = new KafkaProducer<>(properties);
        this.isAsync = isAsync;
        this.topic = topic;
    }

    @Override
    public void run() {
        int num = 1;
        while (true) {
            String msg = "message_" + num;
            //异步发送
            if (isAsync) {
                prod.send(new ProducerRecord<>(topic, num, msg), (recordMetadata, e) -> System.out.println("#offset:" + recordMetadata.offset()));
            }
            //同步发送
            else {
                prod.send(new ProducerRecord<>(topic, num, msg));
            }
        }
    }

    public static void main(String[] args) {
        new Producer("test1", true).start();
        new Producer("test2", false).start();
    }
}
